package com.bingxu.flink.watermark;

import com.bingxu.flink.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * @author :BingXu
 * @description :TODO 迟到数据的处理
 * @date :2021/8/16 19:10
 * @modifier :
 */

public class WMEvent_Custom {
    public static void main(String[] args){
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        env.getConfig().setAutoWatermarkInterval(3000);

        env
                .socketTextStream("localhost", 9999)
                .map((MapFunction<String, WaterSensor>) value -> {
                    String[] fields = value.split(",");
                    return new WaterSensor(fields[0], Long.valueOf(fields[1])*1000, Integer.valueOf(fields[2]));
                })
                // 分配时间戳和水印，参数指定对应的分配策略，静态方法对不同类型的流选择合适的分配器
                .assignTimestampsAndWatermarks(new WatermarkStrategy<WaterSensor>() {
                        @Override
                        public WatermarkGenerator<WaterSensor> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                            return new CustomWaterMark();
                        }
                    }
                    .withTimestampAssigner((element, recordTimestamp) -> element.getTs()))
                .keyBy(ele -> ele.getId())
                // 开窗时间5
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .aggregate(new AggregateFunction<WaterSensor, Long, Long>() {
                    @Override
                    public Long createAccumulator() {
                        return 0L;
                    }

                    @Override
                    public Long add(WaterSensor value, Long accumulator) {

                        return accumulator + value.getVc();
                    }

                    @Override
                    public Long getResult(Long accumulator) {
                        return accumulator;
                    }

                    @Override
                    public Long merge(Long a, Long b) {
                        return null;
                    }
                }, new WindowFunction<Long, String, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window,
                                      Iterable<Long> input, Collector<String> out) throws Exception {
                        Long sum = input.iterator().next();
                        out.collect(s+" "+sum);
                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static class CustomWaterMark implements WatermarkGenerator<WaterSensor>{
        private Long maxStamp = Long.MIN_VALUE+3000;
        /**
         * 事件触发的处理
         * @param event 传入的数据元素
         * @param eventTimestamp 事件时间戳
         * @param output 水印输出对象
         */
        @Override
        public void onEvent(WaterSensor event,
                            long eventTimestamp,
                            WatermarkOutput output) {
            maxStamp = Math.max(event.getTs(), maxStamp);
            // 抛出一个允许迟到3秒的水印,当发射方法在处理方法中时，每一个数据发送一次
            output.emitWatermark(new Watermark(maxStamp-3000));

        }

        /**
         * 发射水印，此方法会被周期性调用
         * @param output
         */
        @Override
        public void onPeriodicEmit(WatermarkOutput output) {
            // 当抛出水印的方法在此方法中，默认是200ms调用一次该方法，抛出水印
//            output.emitWatermark(new Watermark(maxStamp-3000));
        }
    }
}
